package com.atguigu.flink.streamapi.trans;

import com.atguigu.flink.pojo.WordCount;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2023/1/29
 *
 *     数据分发算子，通常都是空参的。
 *          process(KeyedProcessFunction x)
 *          map(MapFunction)
 *
 *          流.分发算子()
 *
 *     用于把数据从上游按照某种规则发送到下游！
 *
 *   -------------------------
 *      ChannelSelector： 接口。通道选择器。下游有多个通道，把数据选择发送到哪个通道。
 *          int selectChannel(T record);  核心方法，计算当前数据应该发到下游的哪个通道(Task)
 *           boolean isBroadcast();   是否是广播。
 *                      单播：  只告诉一个人
 *                      广播:   告诉全世界。 很少用，几乎不用。让数据流成倍的增长！
 *
 *
 *         StreamPartitioner(分区器)
 *           如果希望使用以下的分区器，无需自己创建，只需要调用提供的算子即可。
 *           rescale   RescalePartitioner       在一个TaskManager内部负载均衡。
 *                                      极端情况下，依旧会存在数据倾斜，倾斜不会很明显。
 *
 *           rebalance   RebalancePartitioner    默认。全局负载均衡分发！
 *                              大数据处理中，最害怕数据倾斜
 *           keyBy   KeyGroupStreamPartitioner  .
 *                  key相同，发到下游的同一个Task
 *                  key不同，也有可能发到下游的同一个Task
 *           global   GlobalPartitioner      全局汇总。把数据全部从上游发到下游的 0号 Task
 *           forward   ForwardPartitioner
 *                      上下游的并行度必须是一致的！
 *                      1对1分发。
 *                              map1 ----> print1
 *                              map2 ----> print2
 *                              map3 ----> print3
 *
 *                       如果两个算子之间是forward分发规则，默认会被执行算子链(把两个算子合并为一个Task)。
 *           shuffle   ShufflePartitioner    随机分区
 *           自己继承   CustomPartitionerWrapper(自定义单播)
 *           broadcast 广播  BroadcastPartitioner
 *
 *         在webUI中看到的分发规则是 StreamPartitioner的toString()返回的值。
 *
 *
 */
public class Demo11_Distribute
{
    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",3333);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        env.setParallelism(1);
        //全局禁用算子链
        env.disableOperatorChaining();

        DataStreamSource<String> ds = env.socketTextStream("hadoop103",8888);

        ds.map(line -> line).setParallelism(3)
          .forward()
          .print().setParallelism(3);

        env.execute();

    }
}
